package cn.gwm.flink.streaming.ods;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.DefaultConstant;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.gwm.flink.streaming.dwm.charge.util.ModeCodeUtil;
import cn.gwm.flink.streaming.sink.hbase.HbaseSinkClient;
import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.flink.streaming.task.BaseTask;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.map.LRUMap;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Map;

/**
 * @Author: liangguang
 * @Date: 2022/10/11
 * @Description: ev车  ods处理层 es13
 */
@Slf4j
public class EvOrcTxtPreTaskEs13 {

    private static Integer LONG_SHORT_NUM = 500;
    private static String POSITION_COMMAND = "TBOX_MONITOR";


    public static DataStream<JSONObject> handle(StreamExecutionEnvironment env) {
        String envName = System.getenv().get(DefaultConstant.BOOT_PROFILES_ACTIVE);
        FlinkKafkaConsumer<String> kafkaConsumer = null;
        if ("test".equals(envName)) {
            // aliyun test
            kafkaConsumer = FlinkKafkaUtil.getConsumer(
                    ConfigLoader.get("ali.server"),
                    "consumer-ods-es13-orc",
                    ConfigLoader.get("kafka.topic.es13")
            );
            kafkaConsumer.setStartFromGroupOffsets();
        } else if (envName == null) {
            kafkaConsumer = FlinkKafkaUtil.getConsumerGroupOffsets(
                    "consumer-ods-es13-orc",
                    ConfigLoader.get("kafka.topic.es13")
            );
        }


        // 添加车型字段
        DataStream<JSONObject> jsonSourceStream = env.addSource(kafkaConsumer).map(JSONUtil::parseObj).returns(JSONObject.class)
                .map(item -> {
                    return item.set(VehicleType.VEHICLE_TYPE, VehicleType.ES13)
                            .set(BaseFields.PARTITION_KEY, DateUtil.format(DateUtil.date(item.getLong(BaseFields.tid)), DatePattern.NORM_DATE_PATTERN));
                });

        // 分流
        OutputTag<JSONObject> longStreamTag = new OutputTag<JSONObject>("longStreamTag") {
        };
        OutputTag<JSONObject> locationStreamTag = new OutputTag<JSONObject>("locationStreamTag") {
        };

        SingleOutputStreamOperator<JSONObject> processStream = jsonSourceStream.keyBy(item -> item.getStr(BaseFields.vin))
                .flatMap(new EvOrcTxtPreTaskEs13.Es13DetailFunction())
                .process(new ProcessFunction<JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {
                        if (POSITION_COMMAND.equals(value.getStr(BaseFields.command))) {
                            ctx.output(locationStreamTag, value);
                        } else {
                            ctx.output(longStreamTag, value);
                        }
                    }
                });

        DataStream<JSONObject> longStream = processStream.getSideOutput(longStreamTag);
        DataStream<JSONObject> locationStream = processStream.getSideOutput(locationStreamTag);

        longStream.addSink(BaseTask.getOrcTxtSink(ConfigLoader.get("hive.table.name.ods.es13.long"),
                "odsOrcTxt/",
                BeanSource.SourceEnum.evOdsTh,BeanSource.SourceEnum.evOdsAli));

        locationStream.addSink(BaseTask.getOrcTxtSink(ConfigLoader.get("hive.table.name.ods.es13.location"),
                "odsOrcTxt/",
                BeanSource.SourceEnum.gpsIntranet));

        HbaseSinkClient.hbaseSink(locationStream, ConfigLoader.get("hbase.dwd.table.name.es13.location"), ConfigLoader.get("hbase.table.columnfamily"));

        return longStream;
    }


    public static class Es13DetailFunction extends RichFlatMapFunction<JSONObject, JSONObject> {

        private ValueState<Map> locationState;


        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            StateTtlConfig ttlConfig = StateTtlConfig
                    .newBuilder(Time.seconds(10))
                    .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                    .build();

            ValueStateDescriptor<Map> descriptor = new ValueStateDescriptor<Map>("locationState", Map.class, new LRUMap(16));
            descriptor.enableTimeToLive(ttlConfig);
            locationState = getRuntimeContext().getState(descriptor);
        }

        @Override
        public void flatMap(JSONObject item, Collector<JSONObject> out) throws Exception {
            try {
                if (POSITION_COMMAND.equals(item.getStr("command"))) {
                    // 经纬度数据过滤 和 抽析
                    if (item.getDouble("tboxMoniLatitude") != null && item.getDouble("tboxMoniLongitude") != null) {
                        double latitude = item.getDouble("tboxMoniLatitude").doubleValue();
                        double longitude = item.getDouble("tboxMoniLongitude").doubleValue();
                        if (latitude > 15d && latitude < 55d && longitude > 72d && longitude < 130d) {
                            int second = DateUtil.date(item.getLong(BaseFields.tid)).getField(DateField.SECOND);
                            if ((second & 1) != 1) {
                                out.collect(BeanSource.transJson(item, BeanSource.SourceEnum.evEs13LocationSource));
                            }
                            Map value = locationState.value();
                            if (value == null) {
                                value = new LRUMap(16);
                            }
                            value.put(item.getLong(BaseFields.tid), item);
                            locationState.update(value);
                        }
                    }

                } else {
                    if (item.size() > LONG_SHORT_NUM) {
                        item.set("T_BOX_RefrInfo", item.getStr("T-BOX_RefrInfo"));
                        // 补充车型四码
                        Map<String, String> modeCodeMap = ModeCodeUtil.batteryData(item.getStr(BaseFields.vin));
                        item.set("modeCode", modeCodeMap.get("modelCode"));
                        // 进行经纬度补充
                        Map location = locationState.value();
                        if (location != null) {
                            // 相差不超过3s
                            Long aLong = mistakeMatch(location, item.getLong(BaseFields.tid), 3);
                            if (aLong != null) {
                                JSONObject locationJson = (JSONObject) location.get(aLong);
                                item.set("longitude", locationJson.get("tboxMoniLongitude"));
                                item.set("latitude", locationJson.get("tboxMoniLatitude"));
                            }
                        }
                        out.collect(BeanSource.transJson(item, BeanSource.SourceEnum.evEs13LongSource));
                    }

                }
            } catch (Exception e) {
                e.printStackTrace();
                log.error("EvPreTaskEs13 happen error . msg is " + e.getMessage());
            }

        }
    }


    public static Long mistakeMatch(Map map, Long tid, Integer scope) {
        Long resultKey = null;
        for (int i = 0; i <= scope; i++) {
            Long key = tid - scope * i * 1000;
            if (map.containsKey(key)) {
                resultKey = key;
                break;
            }
            key = tid + scope * i * 1000;
            if (map.containsKey(key)) {
                resultKey = key;
                break;
            }
        }
        return resultKey;
    }

}
